Using Gymnasium Environments

What Gymnasium Controller Does

The Gymnasium Kafka controller wraps OpenAI Gymnasium environments, enabling RL agents to interact via Kafka messages instead of direct Python calls.

Flow:

Environment → Kafka (state) → Agent → Kafka (action) → Environment

Configuration

Basic Configuration

gymnasium:
  # Environment settings
  environment: "Pendulum-v1"
  render_mode: null  # Options: null, 'human', 'rgb_array'
  max_episode_steps: null  # null uses environment default
  
  # Kafka topics
  input_topic: "gymnasium-action"
  output_topics:
    sarsa: "gymnasium-sarsa"
    state: "gymnasium-state"
    decomposed: "gymnasium-output"
  
  # Operation mode
  blocking_mode: false  # true = wait for Kafka actions, false = use default actions
  default_action_strategy: "random"  # Options: 'random', 'zero'
  step_delay: 0.0  # Delay in seconds between steps
  reset_on_start: true  # Reset environment on startup

Key Parameters

environment: Gymnasium environment ID

• Classic: "CartPole-v1", "Pendulum-v1", "MountainCarContinuous-v0"
• MuJoCo: "Ant-v4", "Hopper-v4" (requires license)
• Custom: Your registered environment

blocking_mode:

• false: Use default actions when none received (demo/testing)
• true: Wait for agent actions (Required for RL currently)

default_action_strategy:

• "random": Sample from action space
• "zero": Zero vector/action

step_delay: Seconds between environment steps

• 0.0: Run as fast as possible
• 0.1: 10 steps per second (easier to observe)

Starting Gymnasium Controller

# .env
COMPOSE_PROFILES=gymnasium

docker compose up gymnasium-kafka-controller

Verify Operation

Check logs:

docker compose logs gymnasium-kafka-controller | head -20

Look for:

Created gymnasium environment: Pendulum-v1
Subscribed to topics: ['gymnasium-action']
Starting Kafka Gym wrapper loop

Understanding Output Topics

SARSA Topic

Complete RL transition tuples for training:

{
  "timestamp": 1234567890.123,
  "channels": {
    "state": [0.1, 0.2, -0.5],
    "action": [1.5],
    "reward": -2.3,
    "next_state": [0.15, 0.25, -0.48],
    "done": false,
    "truncated": false
  }
}

Consumer: RL agent data ingest thread

State Topic

Current state only for inference:

{
  "timestamp": 1234567890.123,
  "channels": {
    "state": [0.1, 0.2, -0.5]
  }
}

Consumer: RL agent inference thread

Decomposed Topic

Flattened data for monitoring:

{
  "timestamp": 1234567890.123,
  "channels": {
    "state_0": 0.1,
    "state_1": 0.2,
    "state_2": -0.5,
    "action_0": 1.5,
    "reward": -2.3,
    "episode": 5,
    "episode_step": 23
  }
}

Consumers: InfluxDB, monitoring, autoencoder agents

Blocking vs Non-Blocking Mode

Non-Blocking Mode

Configuration:

blocking_mode: false
default_action_strategy: "random"

Behavior:

• Polls Kafka for actions
• Uses random/zero action if none available
• Continues running even if agent fails

Use when:

• Testing environment setup
• Developing agents
• Fallback behavior desired

Example: Demo runs while RL agent warms up

Blocking Mode (Currently Required for RL)

Configuration:

blocking_mode: true

Behavior:

• Publishes state
• Waits for action from Kafka
• Pauses environment until action received

Use when:

• Agent controls all actions
• Precise step synchronization needed
• No wasted environment steps

Caution: System deadlocks if agent fails as actions are generated by agent and environment waits for agents actions!

Custom Environments

In addition to the traditional Gymnasium environments, we offer the abilty to add custom gymnasium environments as well!

1. Create environment class:

# smocs/control_plane/my_custom_env.py
import gymnasium as gym

class MyCustomEnv(gym.Env):
    def __init__(self):
        self.action_space = gym.spaces.Box(...)
        self.observation_space = gym.spaces.Box(...)
    
    def reset(self):
        # Return initial state
        pass
    
    def step(self, action):
        # Return state, reward, done, truncated, info
        pass

2. Register environment:

# smocs/control_plane/__init__.py
from gymnasium.envs.registration import register

register(
    id='MyCustom-v0',
    entry_point='smocs.control_plane.my_custom_env:MyCustomEnv',
)

3. Configure:

gymnasium:
  environment: "MyCustom-v0"

Episode Management

Episode Lifecycle

Episode start:

Environment resets → Publishes S₀ → Agent generates A₀

Episode running:

Environment receives Aₜ → Executes step → Publishes Sₜ₊₁ and SARSA

Episode end (done or truncated):

Environment resets → New episode begins

Episode Termination

Done: Task completed

• CartPole: Pole fell
• Pendulum: Never (infinite horizon)

Truncated: Time limit reached

• Default: Environment-specific limit
• Override: max_episode_steps in config

gymnasium:
  max_episode_steps: 500  # Override default

Monitoring

Episode Metrics

View in logs:

docker compose logs gymnasium-kafka-controller | grep "Episode.*FINISHED"

Output:

Episode 5 FINISHED
  Total steps: 200
  Total reward: -850.23
  Duration: 20.5s

Message Throughput

Count messages:

docker compose logs gymnasium-kafka-controller | grep "Sent" | wc -l

Check rate:

# Messages per second
docker compose logs --tail=100 gymnasium-kafka-controller | grep "Sent" | wc -l
# Divide by log time range

Common Issues

Agent Not Receiving States

Symptom: Agent logs show no messages

Check Gymnasium is publishing:

docker compose logs gymnasium-kafka-controller | grep "Sent state"

Verify topic:

docker exec kafka-broker kafka-topics.sh --describe --topic gymnasium-state --bootstrap-server localhost:9092

Check agent subscription:

docker compose logs rl-control-agent1 | grep "Subscribed to"

Environment Not Receiving Actions

Symptom: Using default actions instead of agent actions

In non-blocking mode: Expected behavior

Check agent is publishing:

docker compose logs rl-control-agent1 | grep "Sent.*action"

Verify action topic:

# Check messages exist
docker exec kafka-broker kafka-console-consumer.sh \
  --topic gymnasium-action \
  --bootstrap-server localhost:9092 \
  --from-beginning \
  --max-messages 1

Environment Stuck

Symptom: No logs, no progress

In blocking mode: Waiting for action

Solutions:

1. Check agent is running: docker compose ps
2. Verify agent can reach Kafka: Check agent logs
3. Switch to non-blocking for testing:

blocking_mode: false

---